package main.common.java;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.Text;

public class NcdcRecordParser {

	private static final int MISSING_TEMPERATURE = 9999;

	private static final DateFormat DATE_FORMAT = new SimpleDateFormat(
			"yyyyMMddHHmm");

	private String stationId;
	private String observationDateString;
	private String year;
	private String airTemperatureString;
	private int airTemperature;
	private boolean airTemperatureMalformed;
	private String quality;

	public void parse(String record) {
		stationId = record.substring(4, 10) + "-" + record.substring(10, 15);
		observationDateString = record.substring(15, 27);
		year = record.substring(15, 19);
		airTemperatureMalformed = false;
		// Remove leading plus sign as parseInt doesn't like them
		if (record.charAt(87) == '+') {
			airTemperatureString = record.substring(88, 92);
			airTemperature = Integer.parseInt(airTemperatureString);
		} else if (record.charAt(87) == '-') {
			airTemperatureString = record.substring(87, 92);
			airTemperature = Integer.parseInt(airTemperatureString);
		} else {
			airTemperatureMalformed = true;
		}
		airTemperature = Integer.parseInt(airTemperatureString);
		quality = record.substring(92, 93);
	}

	public void parse(Text record) {
		parse(record.toString());
	}

	public boolean isValidTemperature() {
		return !airTemperatureMalformed
				&& airTemperature != MISSING_TEMPERATURE
				&& quality.matches("[01459]");
	}

	public boolean isMalformedTemperature() {
		return airTemperatureMalformed;
	}

	public boolean isMissingTemperature() {
		return airTemperature == MISSING_TEMPERATURE;
	}

	public String getStationId() {
		return stationId;
	}

	public Date getObservationDate() {
		try {
			// System.out.println(observationDateString);
			return DATE_FORMAT.parse(observationDateString);
		} catch (ParseException e) {
			throw new IllegalArgumentException(e);
		}
	}

	public String getYear() {
		return year;
	}

	public int getYearInt() {
		return Integer.parseInt(year);
	}

	public int getAirTemperature() {
		return airTemperature;
	}

	public String getAirTemperatureString() {
		return airTemperatureString;
	}

	public String getQuality() {
		return quality;
	}

	public static void main(String[] args) {
		NcdcRecordParser nrp = new NcdcRecordParser();
		
		File file = new File("/home/zzy/hadoop-0.20.2/input/");
		try {
			Set set = new HashSet();
			File[] files = file.listFiles();
			if (files == null) {
				return;
			}
			for (int i = 0; i < files.length; i++) {

				FileReader fr = new FileReader(files[i]);
				BufferedReader br = new BufferedReader(fr);
				String str = null;
				while ((str = br.readLine()) != null) {
					nrp.parse(str);
					set.add(nrp.getStationId()
							+ nrp.getObservationDate().getTime());
					if (!nrp.isValidTemperature()) {
						System.out.println(str);
					}
				}
			}
			System.out.println(set.size());
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
}
